package io.gitee.skyarthur1987.leporidae.delayqueue.autoconfigure;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import io.gitee.skyarthur1987.leporidae.delayqueue.api.DelayQueue;
import io.gitee.skyarthur1987.leporidae.delayqueue.api.DelayQueuePollEventHandler;
import io.gitee.skyarthur1987.leporidae.delayqueue.exception.DefaultDelayQueueConsumerExceptionHandler;
import io.gitee.skyarthur1987.leporidae.delayqueue.core.DelayQueueConsumer;
import io.gitee.skyarthur1987.leporidae.delayqueue.exception.DelayQueueConsumerExceptionHandler;
import io.gitee.skyarthur1987.leporidae.delayqueue.core.DelayQueuePollEventListener;
import io.gitee.skyarthur1987.leporidae.delayqueue.core.DelayQueuePollScheduler;
import io.gitee.skyarthur1987.leporidae.delayqueue.dto.BatchDelayQueueTask;
import io.gitee.skyarthur1987.leporidae.utils.base.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * 延迟队列消费端配置
 * Created by skyarthur on 2024/3/10
 */
@Configuration
@ConditionalOnProperty(name = "leporidae.delay-queue.consumer.enable", havingValue = "true")
@Slf4j
public class DelayQueueConsumerConfig {

    @Bean
    public Consumer<BatchDelayQueueTask> delayQueueConsumer(@Qualifier("zSetDelayQueue") DelayQueue<String> zSetDelayQueue,
                                                            ApplicationEventPublisher eventPublisher) {
        return new DelayQueueConsumer(zSetDelayQueue, eventPublisher);
    }

    @Bean
    public DelayQueuePollScheduler delayQueuePollScheduler(
            @Qualifier("delayQueueConsumer") Consumer<BatchDelayQueueTask> delayQueueConsumer,
            DelayQueuePollSchedulerConfig schedulerConfig,
            @Qualifier("delayQueueTopic2Handler") Map<String, DelayQueuePollEventHandler> delayQueueTopic2Handler,
            @Autowired(required = false) DelayQueueConsumerExceptionHandler exceptionHandler) {

        List<String> delayQueueTopics = Lists.newArrayList(delayQueueTopic2Handler.keySet());
        DelayQueueConsumerExceptionHandler handler = Optional.ofNullable(exceptionHandler)
                .orElse(new DefaultDelayQueueConsumerExceptionHandler());
        log.info("DelayQueuePollScheduler init, scheduler config: {}, delayQueueTopics: {}, " +
                        "exceptionHandler: {}",
                JsonUtils.toJson(schedulerConfig), JsonUtils.toJson(delayQueueTopics), handler.getClass().getName());
        return new DelayQueuePollScheduler(delayQueueConsumer, schedulerConfig, delayQueueTopics, handler);
    }

    @Bean
    @ConfigurationProperties(prefix = "leporidae.delay-queue.poll-executor")
    public DelayQueuePollSchedulerConfig delayQueuePollSchedulerConfig() {
        return new DelayQueuePollSchedulerConfig();
    }

    @Bean
    public Map<String, DelayQueuePollEventHandler> delayQueueTopic2Handler(@Autowired List<DelayQueuePollEventHandler> eventHandlerList) {
        Map<String, DelayQueuePollEventHandler> delayQueueTopic2Handler = Maps.newHashMap();
        for (DelayQueuePollEventHandler handler : eventHandlerList) {
            handler.getSupportedTopics().forEach(t -> delayQueueTopic2Handler.put(t, handler));
        }
        return delayQueueTopic2Handler;
    }

    @Bean
    public DelayQueuePollEventListener delayQueuePollEventListener(
            @Qualifier("delayQueueTopic2Handler") Map<String, DelayQueuePollEventHandler> delayQueueTopic2Handler) {
        log.info("DelayQueuePollEventListener init, delayQueueTopic2Handler: {}",
                delayQueueTopic2Handler.keySet().stream()
                        .map(x -> String.format("%s:%s", x, delayQueueTopic2Handler.get(x).getClass().getName()))
                        .collect(Collectors.joining(",")));
        return new DelayQueuePollEventListener(delayQueueTopic2Handler);
    }
}
